package njuflink2;

import njuflink.GlobalConfig;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.Statement;
import java.util.Properties;

public class KafkaSourceMySQL2 {
        public static void main(String[] args) throws Exception{
            Properties prop = new Properties();
            //指定kafka broker地址
//            prop.put("bootstrap.servers", "172.19.240.61:9092");
            prop.put("bootstrap.servers", njuflink2.GlobalConfig.BOOTSRAP_SEVERS );
            //指定key value的序列化方式
            prop.put("key.serializer", StringSerializer.class.getName());
            prop.put("value.serializer", StringSerializer.class.getName());
            //指定topic名称
            String topic = njuflink2.GlobalConfig.TOPIC2;
            //创建producer链接
            KafkaProducer<String, String> producer = new KafkaProducer<String,String>(prop);
            System.out.println("connecting mysql");
            Class.forName(njuflink2.GlobalConfig.DRIVER_CLASS);
            String url = njuflink2.GlobalConfig.DB_URL;
            String username= njuflink2.GlobalConfig.USER_NAME;
            String password= GlobalConfig.PASSWORD;
            Connection connection = DriverManager.getConnection(url,username,password);
            Statement statement = connection.createStatement();
            int offset=0,rows=100;
            ResultSet resultSet;
            while(true){
                String sql = "SELECT * FROM repo_info limit "+offset+","+rows;
                offset+=rows;
                resultSet = statement.executeQuery(sql);
                String line = null;
                if(resultSet==null){
                    break;
                }
                while (resultSet.next()) {
                    line="";
                    String tmplang=resultSet.getString("language");
                    String tmplic=resultSet.getString("license");
                    if(tmplang==null||tmplang.length()>50){
                        line+="Unknown Language";
                    }
                    else{
                        line+=resultSet.getString("language");
                    }
                    line+=",";
                    if(tmplic==null||tmplic.length()>50){
                        line+="Unknown License";
                    }
                    else{
                        line+=resultSet.getString("License");
                    }
                    //line = resultSet.getString("language")+","+resultSet.getString("license");
                    if(line==null||line==""){
                        continue;
                    }
                    System.out.println(line);
                    //生产消息
                    producer.send(new ProducerRecord<String, String>(topic,line));
                    //Thread.sleep(10);
                }
                Thread.sleep(1000);
                System.out.println("wait for 1 s");
            }
//            String sql = "SELECT *FROM repo_info ;";
//            ResultSet resultSet = statement.executeQuery(sql);
//            String line = null;
//            while (resultSet.next()) {
//
//                line = resultSet.getString("language");
//                if(line==null||line==""){
//                    continue;
//                }
//                System.out.println(line);
//                //生产消息
//                producer.send(new ProducerRecord<String, String>(topic,line));
//                Thread.sleep(10);
//            }
            resultSet.close();
            statement.close();
            connection.close();
            //关闭链接
            producer.close();
    }

}